package com.codejiwei.flink.practice;

import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @author jiwei
 * @description 1.两个流中存储的数据类型可以不同
 * 2.只是机械的合并在一起, 内部仍然是分离的2个流
 * 3.只能2个流进行connect, 不能有第3个参与
 * @date 2023/5/19 17:21
 */
public class Flink_Transform_Connect {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);
        DataStreamSource<Integer> intStream = env.fromElements(1, 2, 3, 4, 5);
        DataStreamSource<String> stringStream = env.fromElements("a", "b", "c");

        ConnectedStreams<Integer, String> connectedStreams = intStream.connect(stringStream);

        connectedStreams.getFirstInput().print("first");
        connectedStreams.getSecondInput().print("second");

        System.out.println(connectedStreams.getType1().toString());
        System.out.println(connectedStreams.getType2().toString());

        env.execute();
    }
}
